/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2022 All Rights Reserved.
 */
package com.example.spider.proxy.core.distinct;

import jakarta.annotation.Nonnull;
import jakarta.annotation.Nullable;
import lombok.extern.slf4j.Slf4j;
import org.redisson.RedissonLocalCachedMap;
import org.shoulder.core.lock.LockInfo;
import org.shoulder.core.lock.ServerLock;

import java.io.Serializable;
import java.time.Duration;
import java.util.*;

/**
 * 写前拿锁，与存储同步（redis）
 * 读则只读内存
 * 适合总量比较小的（如 key<1w）且读多写很少，而且允许多个进程节点写入，只增不减，谁都有权写入。key多了占用内存多
 *
 * @see RedissonLocalCachedMap
 * key = 9909
 * value = 2
 *
 * @author lym
 */
@Slf4j
public class LockOnWriteCacheMap<KEY extends Serializable, VALUE_TYPE extends Serializable> implements Map<KEY, VALUE_TYPE> {

    private final String keyPrefix;

    private volatile Map<KEY, VALUE_TYPE> memoryCache;

    private final ServerLock lock;

    private final Map<KEY, VALUE_TYPE> globalUniqueStorage;

    public LockOnWriteCacheMap(String keyPrefix, Map<KEY, VALUE_TYPE> globalUniqueStorage, ServerLock lock) {
        this.keyPrefix = keyPrefix;
        this.memoryCache = new HashMap<>(64);
        this.globalUniqueStorage = globalUniqueStorage;
        this.lock = lock;
    }

    @Override public int size() {
        return memoryCache.size();
    }

    @Override public boolean isEmpty() {
        return memoryCache.isEmpty();
    }

    @Override public boolean containsKey(Object key) {
        return memoryCache.containsKey(key);
    }

    @Override public boolean containsValue(Object value) {
        return memoryCache.containsValue(value);
    }

    @Override public VALUE_TYPE get(Object key) {
        return memoryCache.get(key);
    }

    @Nullable @Override public VALUE_TYPE put(KEY key, VALUE_TYPE value) {
        throw new UnsupportedOperationException("not support put");
    }

    @Nullable @Override public VALUE_TYPE putIfAbsent(KEY key, VALUE_TYPE value) {
        if(memoryCache.get(key) != null) {
            return putIfAbsent(nextKey(key), value);
        }
        LockInfo lockInfo = lock(key);
        VALUE_TYPE old;
        try {
            // reload to memoryCache
            reloadToMemoryCache();
            if ((old = memoryCache.get(key)).equals(value)) {
                // 相当于本地也 put 成功了
                return null;
            }
            if(old == null) {
                // 没有，插入

            }
            // 使用者考虑如何操作
            KEY nextKey = nextKey(key);
            if (Objects.equals(key, nextKey)) {
                // 不换key
                return old;
            }
            // fixme unlock 换key 重试
            return putIfAbsent(nextKey, value);
        } finally {
            unlock(lockInfo);
        }
    }

    protected synchronized KEY nextKey(KEY key) {
        return key;
    }

    private synchronized void reloadToMemoryCache() {
        this.memoryCache = new HashMap<>(globalUniqueStorage);
    }

    protected LockInfo lock(KEY key) {
        LockInfo lockInfo = new LockInfo(keyPrefix + ":" + key, Duration.ofSeconds(10));
        lock.lock(lockInfo);
        return lockInfo;
    }

    protected void unlock(LockInfo lockInfo) {
        lock.unlock(lockInfo);
    }

    @Override public VALUE_TYPE remove(Object key) {
        throw new UnsupportedOperationException("not support remove");
    }

    @Override
    public void putAll(@Nonnull Map<? extends KEY, ? extends VALUE_TYPE> m) {
        throw new UnsupportedOperationException("not support remove");
    }

    @Override public void clear() {
        throw new UnsupportedOperationException("not support clear");
    }

    @Nonnull
    @Override
    public Set<KEY> keySet() {
        return memoryCache.keySet();
    }

    @Nonnull
    @Override
    public Collection<VALUE_TYPE> values() {
        return memoryCache.values();
    }

    @Nonnull
    @Override
    public Set<Entry<KEY, VALUE_TYPE>> entrySet() {
        return memoryCache.entrySet();
    }
}